package spring.cloud.tasks.client.item_task_processor;

import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.client.ItemTaskWorker;
import spring.cloud.tasks.client.item_task_processor.task_item.ServerTaskItemsSnapshotInfo;
import spring.cloud.tasks.common.IdentityThread;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class TaskItemsReAssignThread extends IdentityThread {
    private ItemTaskWorker itemTaskWorker;
    private String workerId;
    private ReentrantLock lock = new ReentrantLock();

    public TaskItemsReAssignThread(String threadId, ItemTaskWorker itemTaskWorker) {
        super(threadId);
        this.itemTaskWorker = itemTaskWorker;
        this.workerId = itemTaskWorker.getWorkerId();
    }

    @Override
    public void run() {
        boolean tryLock = false;
        try {
            tryLock = lock.tryLock(5, TimeUnit.SECONDS);
            synchronized (TaskItemsReAssignThread.class) {
                synchronized (itemTaskWorker) {//加锁执行
                    cleanTaskItemsIfNecessary();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (tryLock) {
                lock.unlock();
            }

        }

    }

    private void cleanTaskItemsIfNecessary() {
        //获取所有的任务 遍历所有的任务项
        List<String> taskIdList = itemTaskWorker.getTaskIdList();
        Map<String, List<TaskProcessor>> taskIdToTaskProcessorListMap = itemTaskWorker.getTaskIdToTaskProcessorListMap();
        for (String taskId : taskIdList) {
            if (!taskIdToTaskProcessorListMap.containsKey(taskId)) {
                //什么都不需要做
            } else {
                try {
                    cleanTaskItemsIfNecessary(taskId, taskIdToTaskProcessorListMap);
                } catch (Exception e) {
                    log.error("cleanTaskItemsIfNecessary:" + taskId, e);
                }
            }
        }
    }

    /**
     * 只是清理 不进行拉取
     */
    private void cleanTaskItemsIfNecessary(String taskId, Map<String, List<TaskProcessor>> taskIdToTaskProcessorListMap) throws Exception {
        Task task = itemTaskWorker.getZooKeeperItemTaskDataManager().loadTask(taskId);
        if (task == null) {
            return;
        }
        ServerTaskItemsSnapshotInfo serverTaskItemsSnapshotInfo = itemTaskWorker.getZooKeeperItemTaskDataManager().getServerTaskItemsSnapshotInfo(taskId);
        if (serverTaskItemsSnapshotInfo == null || ServerTaskItemsSnapshotInfo.task_items_status_initing.equals(serverTaskItemsSnapshotInfo.getStatus())) {
            return;
        }
        //比较本地和server的版本
        Map<String, Integer> taskIdToTaskItemVersionMap = itemTaskWorker.getTaskIdToTaskItemVersionMap();
        Integer taskItemVersion = taskIdToTaskItemVersionMap.get(taskId);
        if (taskItemVersion == null) {
            return;
        }
        int $taskItemVersion = serverTaskItemsSnapshotInfo.getVersion();
        if (taskItemVersion < $taskItemVersion) {
            //需要清空本地的taskItems
            List<TaskProcessor> taskProcessorList = taskIdToTaskProcessorListMap.get(taskId);
            if (taskProcessorList != null) {
                for (TaskProcessor taskProcessor : taskProcessorList) {
                    if (taskProcessor != null) {
                        try {
                            taskProcessor.clearTaskItemAndClearTaskDataIdList();
                        } catch (Exception e) {
                            log.warn("", e);
                        }
                    }
                }
            }
        }

    }
}
